#include <algorithm>
#include <cppformat/format.h>

#include "client.hpp"
#include "proxy.hpp"
#include "server.hpp"
#include "except/exceptions.hpp"
#include "utils/logging.hpp"
#include "syscalls/poll.h"

using namespace cerb;

static msize_t const MAX_PIPE = 64;
static msize_t const MAX_RESPONSES = 256;

Client::Client(int fd, Proxy* p)
    : ProxyConnection(fd)
    , _proxy(p)
    , _awaiting_count(0)
{
    LOG(DEBUG) << "Create Client " << fd;
    p->poll_add_ro(this);
}

Client::~Client()
{
    for (Server* svr: _peers) {
        svr->pop_client(this);
    }
    _proxy->pop_client(this);
}

void Client::on_events(int events)
{
    if (poll::event_is_hup(events)) {
        return close();
    }
    try {
        if (poll::event_is_read(events)) {
			LOG(DEBUG) << "cfd " << fd << " readable";
            _read_request();
        }
        if (closed()) {
            return;
        }
        if (poll::event_is_write(events)) {
			LOG(DEBUG) << "cfd " << fd << " writable";
            _write_response();
        }
        if (_output_buffer_set.empty()) {
			LOG(TRACE) << "_output_buffer_set empty, set " << str() << " || false";
            _proxy->set_conn_poll_ro(this);
        } else {
			LOG(TRACE) << "_output_buffer_set non-empty, set " << str() << " writable";
            _proxy->set_conn_poll_rw(this);
        }
    } catch (BadRedisMessage& e) {
        LOG(DEBUG) << fmt::format("Receive bad message from {} because {}", str(), e.what());
        LOG(DEBUG) << "Dump buffer " << _buffer.to_string();
        return close();
    } catch (IOErrorBase& e) {
        LOG(DEBUG) << "IOError: " << e.what() << " :: Close " << str();
        return close();
    }
}

void Client::after_events(std::set<Connection*>&)
{
    if (closed()) {
		LOG(DEBUG) << "Client closed, delete it";
        delete this;
    }
}

std::string Client::str() const
{
    return fmt::format("Client({})", fd);
}

void Client::_send_buffer_set()
{
    if (_output_buffer_set.writev(fd)) {
		LOG(TRACE) << "_output_buffer_set writev ok";
        for (auto const& g: _ready_groups) {
            g->collect_stats(_proxy);
        }
        _ready_groups.clear();
        _peers.clear();
		LOG(TRACE) << "clear _ready_groups,_peers";
        if (!_parsed_groups.empty()) {
			LOG(TRACE) << "_parsed_groups non-empty";
            _process();
        }
    }
}

//这个是在Server收到响应后通知Client来调用的
void Client::_push_awaitings_to_ready()
{
    if (_awaiting_count != 0 || (!_ready_groups.empty() &&
            _awaiting_groups.size() + _ready_groups.empty() > MAX_RESPONSES))
    {
		LOG(TRACE) << "";
        return;
    }
    for (util::sptr<CommandGroup>& g: _awaiting_groups) {
        g->append_buffer_to(_output_buffer_set);//往_output_buffer_set里加入g的command
        _ready_groups.push_back(std::move(g));
		LOG(TRACE) << "_ready_groups push_back";
    }
	LOG(TRACE) << "_awaiting_groups cleared";
    _awaiting_groups.clear();
    if (!_output_buffer_set.empty()) {
		LOG(TRACE) << "_output_buffer_set non-empty,set rw for _conn_poll_type::" << str();
        _proxy->set_conn_poll_rw(this);
    }
}

void Client::_write_response()
{
    if (!_output_buffer_set.empty()) {
		LOG(TRACE) << "_output_buffer_set non-empty";
        _send_buffer_set();
    }
    if (_awaiting_groups.empty() || _awaiting_count != 0) {
		LOG(TRACE) << "_awaiting_groups empty or _awaiting_count!=0";
        return;
    }
    _push_awaitings_to_ready();
    _send_buffer_set();
}

void Client::_read_request()
{
    int n = _buffer.read(fd);
    LOG(DEBUG) << "Read from " << str() << " current buffer size: "
               << _buffer.size() << " read returns " << n;
    if (n == 0) {
        return close();
    }
	LOG(TRACE) << "read content " << _buffer.to_string();

	//buffer按引用传递，返回时会被清
    ::split_client_command(_buffer, util::mkref(*this));
	LOG(DEBUG) << "client command parse over,buffer cleared ";
    if (_awaiting_groups.empty()) {
		LOG(TRACE) << "_awaiting_groups empty, so call _process";
        _process();
    }
}

void Client::reactivate(util::sref<Command> cmd)
{
    Server* s = cmd->select_server(_proxy);
    if (s == nullptr) {
        return;
    }
    LOG(DEBUG) << "reactivated " << s->str();
    _proxy->set_conn_poll_rw(s);
}

void Client::_process()
{
    msize_t pipe_groups = std::min(msize_t(_parsed_groups.size()), MAX_PIPE);
    LOG(DEBUG) << fmt::format("{} Process {} commands", str(), _parsed_groups.size());
    for (msize_t i = 0; i < pipe_groups; ++i) {
        auto& g = _parsed_groups[i];//这里为何用引用，指向指针？
        if (g->long_connection()) {
            _proxy->poll_del(this);
            g->deliver_client(_proxy);
            LOG(DEBUG) << "Convert self to long connection, close " << str();
            return close();
        }

        if (g->wait_remote()) {
			//only StatsCommandGroup need to wait
            ++_awaiting_count;
            g->select_remote(_proxy);
        }
		LOG(TRACE) << "push _parsed_groups to _awaiting_groups";
        _awaiting_groups.push_back(std::move(g));
    }
    if (pipe_groups == _parsed_groups.size()) {
		LOG(TRACE) << "clear _parsed_groups";
        _parsed_groups.clear();
    } else {
		LOG(TRACE) << "erase _parsed_groups";
        _parsed_groups.erase(_parsed_groups.begin(),_parsed_groups.begin() + pipe_groups);
    }

    if (0 < _awaiting_count) {
		//LOG(TRACE) << "peers size " << _peers.size();
        for (Server* svr: _peers) {
            _proxy->set_conn_poll_rw(svr);
        }
    } else {
        _push_awaitings_to_ready();
    }
}

void Client::group_responsed()
{
    --_awaiting_count;
	LOG(TRACE) << "_awaiting_count decr(" << _awaiting_count << "),push await to ready";
	_push_awaitings_to_ready();
}

void Client::add_peer(Server* svr)
{
	LOG(TRACE) << str() << " add peer " << svr->addr.str();
    _peers.insert(svr);
}

void Client::push_command(util::sptr<CommandGroup> g)
{
    _parsed_groups.push_back(std::move(g));
	LOG(TRACE) << "_parsed_groups size=" << _parsed_groups.size();
}
